package com.iwhalecloud.uncc.mq.listener;

import com.ctg.mq.api.bean.MQResult;
import com.ctg.mq.api.listener.ConsumerQueueListener;
import com.ctg.mq.api.listener.ConsumerQueueStatus;
import com.iwhalecloud.uncc.utils.ZipUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

@Slf4j
public class MqQueueListener extends MqListener  implements MessageListenerOrderly, ConsumerQueueListener {

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
        boolean flag = true;
        try{
            if(list != null && !list.isEmpty()){
//                String message = new String(ZipUtil.uzip(list.get(0).getBody()), StandardCharsets.UTF_8);
                flag = this.handler.dealMessage(list);
            }
        }catch (Exception e){
            log.error("Consume Message Error! Handler is " + this.handler.getClass().getName() + ", errorMsg:" +
                    ExceptionUtils.getStackTrace(e));
            flag = false;
        }
        if(flag){
            return ConsumeOrderlyStatus.SUCCESS;
        }else{
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    }

    @Override
    public ConsumerQueueStatus onMessage(List<MQResult> list) {
        boolean flag = true;
        try{
            if(list != null && !list.isEmpty()){
//                String message = new String(ZipUtil.uzip(list.get(0).getMessage().getBody()), StandardCharsets.UTF_8);
                flag = this.handler.dealMessage(list);
            }
        }catch (Exception e){
            log.error("Consume Message Error! Handler is " + this.handler.getClass().getName() + ", errorMsg:" +
                    ExceptionUtils.getStackTrace(e));
            flag = false;
        }
        if(flag){
            return ConsumerQueueStatus.SUCCESS;
        }else{
            return ConsumerQueueStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    }

}
